package org.lds.simple.disruptor;

import org.lds.simple.disruptor.sequence.Sequence;
import org.lds.simple.disruptor.strategy.WaitStrategy;

public class SequenceBarrier {
    private final int bufferSize;
    private final Sequence producerSequence;
    private final Sequence[] consumerSequence;

    private final WaitStrategy waitStrategy;

    public SequenceBarrier(int bufferSize, Sequence producerSequence, Sequence[] consumerSequence, WaitStrategy waitStrategy) {
        this.bufferSize = bufferSize;
        this.producerSequence = producerSequence;
        this.consumerSequence = consumerSequence;
        this.waitStrategy = waitStrategy;
    }

    private long getCurrProducerSequence() {
        return producerSequence.get();
    }

    public long getCurrMinConsumerSequence() {
        long minConsumerSequence = consumerSequence[0].get();
        for (int i = 1, length = consumerSequence.length; i < length; i++) {
            minConsumerSequence = Math.min(minConsumerSequence, consumerSequence[i].get());
        }
        return minConsumerSequence;
    }

    public long consumerWait(Sequence producerSequence, Sequence consumerSequence) {
        return waitStrategy.consumerWait(producerSequence, consumerSequence);
    }

    public void notifyConsumer() {
        waitStrategy.notifyConsumer();
    }

    public long allowProduce(long currProducerSequence, long currMinConsumerSequence) {
        long nextProducerSequence = currProducerSequence + 1;
        if (nextProducerSequence > currMinConsumerSequence && (nextProducerSequence - currMinConsumerSequence) <= bufferSize) {
            return nextProducerSequence;
        }

        return -1L;
    }

    public long allowProduce() {
        long nextProducerSequence = getCurrProducerSequence() + 1;
        long currMinConsumerSequence = getCurrMinConsumerSequence();
        if (nextProducerSequence > currMinConsumerSequence && (nextProducerSequence - currMinConsumerSequence) <= bufferSize) {
            return nextProducerSequence;
        }

        return -1L;
    }

}
